package com.shujia.traffic.dws

import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.expressions.Window

object DwsAppAcdMrjqDay {
  def main(args: Array[String]): Unit = {
    // 构建SparkSession环境
    val spark = SparkSession
      .builder()
      .appName("DwsAppAcdMrjqDay")
      .enableHiveSupport() // 开启Hive的支持
      .config("spark.sql.shuffle.partitions", "1")
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    val pt = "20250610"
    // 构建DataFrame，再基于DF通过DSL完成数据分析任务
    val acdDF = spark.table("dwd.dwd_base_acd_file_msk_d").where($"pt" === pt)

    var drDF = acdDF
      .groupBy(substring($"sgfssj", 1, 10) as "tjrq")
      .agg(
        count($"sgbh") as "dr_sgs"
        , sum(when($"swrs" + $"swrs24" + $"swrs3" + $"swrs7" + $"swrs30" + $"swrsq" > 0, 1).otherwise(0)) as "dr_swsgs"
        , sum($"swrs" + $"swrs24" + $"swrs3" + $"swrs7" + $"swrs30" + $"swrsq") as "dr_swrs"
        , sum($"ssrs" + $"ssrs24" + $"ssrs3" + $"ssrs7" + $"ssrs30") as "dr_ssrs"
        , sum($"zjccss") as "dr_zjccss"
      )

    val indexList = List("sgs","swsgs","swrs","ssrs","zjccss")
    indexList.foreach(index=>{
      drDF = drDF
        // 计算今年的总数
        .withColumn(s"jn_$index", sum($"dr_$index") over Window.partitionBy(year($"tjrq")))
        // 取去年同期的
        .withColumn(s"last_dr_$index", lag($"dr_$index", 1) over Window.partitionBy(substring($"tjrq", 6, 5)).orderBy(year($"tjrq")))
        // 计算同比
        .withColumn(s"tb_$index",coalesce(round(($"dr_$index"/$"last_dr_$index" - 1) * 100,2),expr("0")))
    })

    // 整理数据并保存到结果目录
    val resDF = drDF
      .select(
        $"tjrq"
        , $"dr_sgs"
        , $"jn_sgs"
        , $"tb_sgs"
        , $"dr_swsgs"
        , $"jn_swsgs"
        , $"tb_swsgs"
        , $"dr_swrs"
        , $"jn_swrs"
        , $"tb_swrs"
        , $"dr_ssrs"
        , $"jn_ssrs"
        , $"tb_ssrs"
        , $"dr_zjccss"
        , $"jn_zjccss"
        , $"tb_zjccss"
      )

    // 增加写入的分区
    spark.sql(
      s"""
        |alter table dws.dws_app_acd_mrjqi_d add if not exists partition (pt=$pt)
        |""".stripMargin)

    resDF
      .write
      .format("csv")
      .option("sep","$")
      .mode(SaveMode.Overwrite)
      .save(s"/project/traffic/dws/dws_app_acd_mrjqi_d/pt=$pt")
  }

  /**
   * 1、打包并上传到服务器
   * 2、通过spark-submit提交任务
   *  spark-submit --master local --class com.shujia.traffic.dws.DwsAppAcdMrjqDay traffic_szxy-1.0-SNAPSHOT.jar
   */
}
